package com.gmscosmo.janus.sendMessage;

import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.gmscosmo.model.Node;
import com.gmscosmo.janus.util.JanusUtil;
import com.gmscosmo.janus.queue.RabbitMqUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;

/**
 * ClassName: AsyncSender
 * Description:
 * Date: 2021/10/12 16:24
 *
 * @author stephen.qiu
 * @version 1.0
 */
@Component
public class JanusAsyncSender {
    @Lazy
    @Autowired
    JanusUtil janusUtil;

    @Lazy
    @Autowired
    RabbitMqUtil rabbitMqUtil;

    private static final CopyOnWriteArraySet<MessageObserver.Observer> mObservers = new CopyOnWriteArraySet<>();

    public interface MessageObserver {
        interface Observer {
            void on(String transaction, JSONObject data);
        }
    }

    /**
     * @param transaction 回复事务号
     * @param ack         是否需要确认信息
     * @param body        信息
     * @return
     * @throws JSONException
     */
    public Future<JSONObject> sendWithFuture(String transaction, boolean ack, String body, int nodeId) throws JSONException {
        janusUtil.addTransaction(transaction);
        Node node = janusUtil.getNodeInfo(nodeId);
        return new ActionCallBack(transaction, ack, node).sendAckCallback(mObservers, rabbitMqUtil, body);
    }

    public static void notifyObservers(String transaction, JSONObject data) {
        for (final MessageObserver.Observer observer : mObservers) {
            observer.on(transaction, data);
        }
    }

    private class ActionCallBack {
        private final String callbackTransaction;
        private final boolean callbackAck;
        private JSONObject response;
        private final Node sendNode;

        ActionCallBack(String transaction, boolean ack, Node node) {
            callbackTransaction = transaction;
            callbackAck = ack;
            sendNode = node;
        }

        @Async("ThreadPoolTaskExecutor")
        public Future<JSONObject> sendAckCallback(CopyOnWriteArraySet<MessageObserver.Observer> mObservers, RabbitMqUtil rabbitMqUtil, String body) {
            CountDownLatch countDownLatch = new CountDownLatch(1);
            MessageObserver.Observer observer = (transaction, data) -> {
                if ("ack".equals(data.getString("janus"))) {
                    if (callbackAck) {
                        if (transaction.equals(callbackTransaction)) {
                            response = data;
                            countDownLatch.countDown();
                        }
                    }
                } else {
                    if (transaction.equals(callbackTransaction)) {
                        response = data;
                        countDownLatch.countDown();
                    }
                }
            };
            mObservers.add(observer);
            rabbitMqUtil.sendMq(sendNode.getTo_janus(), sendNode.getTo_janus(), BuiltinExchangeType.DIRECT, body);
            try {
                countDownLatch.await(5,TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                countDownLatch.countDown();
            }
            mObservers.remove(observer);
            return new AsyncResult<>(response);
        }
    }
}
